package com.zsd.android.rxjavalib;

public class ObservableMap<T, R> implements ObservableOnSubscribe<R> {

    private ObservableOnSubscribe<T> source;
    private Function<? super T, ? extends R> function;
    private Observer<? super R> observableEmitter;

    public ObservableMap(ObservableOnSubscribe<T> source, Function<? super T, ? extends R> function) {
        this.source = source;
        this.function = function;
    }

    @Override
    public void subscribe(Observer<? super R> observableEmitter) {
        this.observableEmitter = observableEmitter;
        MapObserver<T> mapObserver = new MapObserver(observableEmitter, source, function);
        source.subscribe(mapObserver);
    }

    class MapObserver<T> implements Observer<T> {

        private Observer<? super R> observableEmitter;
        private ObservableOnSubscribe<T> source;
        private Function<? super T, ? extends R> function;

        public MapObserver(Observer<? super R> observableEmitter, ObservableOnSubscribe<T> source, Function<? super T, ? extends R> function) {
            this.observableEmitter = observableEmitter;
            this.source = source;
            this.function = function;
        }

        @Override
        public void onSubscribe() {
            observableEmitter.onSubscribe();
        }

        @Override
        public void onNext(T item) {
            try {
                R r = function.apply(item);
                observableEmitter.onNext(r);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onError(Throwable e) {
            observableEmitter.onError(e);
        }

        @Override
        public void onComplete() {
            observableEmitter.onComplete();
        }
    }
}
